import akshare as ak
import asyncio
import queue
import threading
import time
import json
import aiomysql


class MovingAverageStrategy:
    def __init__(self, symbol):
        self.symbol = symbol
        self.kline_data = []
        self.ma_period = 5

    def update_kline_data(self, new_data):
        self.kline_data.append(new_data)
        if len(self.kline_data) > self.ma_period:
            self.kline_data.pop(0)  # Keep only the last 'ma_period' bars

    def calculate_moving_average(self):
        if len(self.kline_data) < self.ma_period:
            return None  # Not enough data to calculate the moving average
        close_prices = [bar['close'] for bar in self.kline_data]
        return sum(close_prices) / len(close_prices)

    def generate_signal(self, latest_bar, external_indicator):
        ma = self.calculate_moving_average()
        if ma is not None:
            combined_value = ma + external_indicator
            if latest_bar['close'] > combined_value:
                return "Buy Signal"
            elif latest_bar['close'] < combined_value:
                return "Sell Signal"
        return None  # No signal


async def simulate_data_feed(symbol, kline_queue, num_init_bars=10):
    # Simulate receiving minute-level K-line data from akshare
    try:
        # Fetch historical data
        historical_data = ak.futures_zh_minute_sina(symbol=symbol, period="1")
        historical_data['symbol'] = symbol
        historical_data = historical_data.tail(num_init_bars).to_dict(orient='records')
        for bar in historical_data:
            kline_queue.put(bar)
            await asyncio.sleep(1)

        # Continuously fetch real-time data
        while True:
            # Fetch recent minute-level data using akshare
            kline_data = ak.futures_zh_minute_sina(symbol=symbol, period="1")
            kline_data['symbol'] = symbol
            if not kline_data.empty:
                latest_bar = kline_data.iloc[-1].to_dict()
                kline_queue.put(latest_bar)
                await asyncio.sleep(60)  # Adjust the sleep time based on your data frequency

    except Exception as e:
        print(f"Error fetching data: {e}")


async def send_to_mysql(pool, signal, bar):
    # Insert signal into MySQL database
    async with pool.acquire() as conn:
        async with conn.cursor() as cursor:
            await cursor.execute(
                "INSERT INTO t_strategy_data (strategy_id,message,open_price,close_price,datetime) VALUES (%s,%s,%s,%s,%s)", (bar['symbol'], signal, bar['open'], bar['close'], bar['datetime']))
        await conn.commit()


def load_mysql_config_from_json(file_path):
    with open(file_path) as f:
        return json.load(f)


async def main():
    symbol = 'rb2401'  # Replace with your desired futures contract symbol
    kline_queue = queue.Queue()

    # Load MySQL configuration from JSON file
    mysql_config = load_mysql_config_from_json('mysql_config.json')

    # Start data simulation task
    await simulate_data_feed(symbol, kline_queue, num_init_bars=10)

    # Initialize the strategy
    strategy = MovingAverageStrategy(symbol)

    # Initialize MySQL connection pool
    async def create_pool():
        return await aiomysql.create_pool(**mysql_config)

    pool = asyncio.run(create_pool())

    while True:
        try:
            # Get the latest K-line data from the queue
            latest_bar = kline_queue.get_nowait()

            # Update the strategy with the latest data
            strategy.update_kline_data(latest_bar)

            # Your external indicator logic here (replace 0 with your actual external indicator value)
            external_indicator = 0

            # Generate signals and print
            signal = strategy.generate_signal(latest_bar, external_indicator)
            if signal:
                print(f"Received Signal: {signal}")

                # Record signal in MySQL asynchronously
                asyncio.run(send_to_mysql(pool, signal, latest_bar))

        except queue.Empty:
            pass  # The queue is empty, continue waiting

        time.sleep(1)  # Adjust the sleep time based on your actual data frequency


if __name__ == "__main__":
    asyncio.run(main())
